package com.wpp.boot;

import com.wpp.client.RPCConsumer;
import com.wpp.coder.RpcEncoder;
import com.wpp.constant.Constant;
import com.wpp.handler.UserClientHandler;
import com.wpp.protocol.RpcRequest;
import com.wpp.register.ServerInfo;
import com.wpp.register.ZkRegisterCenter;
import com.wpp.register.metric.ResponseTimeMetricHolder;
import com.wpp.register.metric.TimeMetric;
import com.wpp.serializer.JSONSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public class ConsumerBootstrap {

    private ZkRegisterCenter registerCenter;

    public ConsumerBootstrap(ZkRegisterCenter registerCenter) {
        this.registerCenter = registerCenter;
    }
    // 客户端启动起到对象
//    private static Bootstrap bootstrap = new Bootstrap();

    //编写方法，初始化客户端（创建连接池 bootstrap 设置bootstrap 连接服务器）
    public void initClient(ServerInfo serverInfo) throws InterruptedException {
        // 创建连接池对象
        NioEventLoopGroup group = new NioEventLoopGroup();
        JSONSerializer jsonSerializer = new JSONSerializer();
//        // 客户端启动起到对象
        Bootstrap bootstrap = new Bootstrap();
        // 对引导对象进行配置
        bootstrap.group(group)
                // 设置通道为nio
                .channel(NioSocketChannel.class)
                // 设置请求协议为tcp
                .option(ChannelOption.TCP_NODELAY,true)
                // 监听channel 并初始化
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        // 获取ChannelPipeline
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        // 这是编码
                        pipeline.addLast(new RpcEncoder(RpcRequest.class,jsonSerializer));
                        pipeline.addLast(new StringDecoder())
                                // todo  这里是否 可以多个连接 一个 handler呢  待校验?
                                .addLast(new UserClientHandler());
                    }
                });
        ChannelFuture channelFuture = bootstrap.connect(serverInfo.getIp(), serverInfo.getPort()).sync();
        Channel channel = channelFuture.channel();
        serverInfo.setChannel(channel);
        channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                // 建立起连接后 对 该服务进行监听
                if (future.isSuccess()) {
                    String serverStr = new StringBuilder(serverInfo.getIp()).append(":").append(serverInfo.getPort()).toString();
                    startTimeWatch(serverStr);
                }
            }
        });
        System.out.println("连接上了服务： "+serverInfo);
    }

    /**
     *  定时器 每5秒 上报一次
     * @param serviceAddressNode
     */
    public void startTimeWatch(String serviceAddressNode) {
        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);

        // 开启定时器
        scheduledThreadPoolExecutor.scheduleWithFixedDelay(()->{
            // 每5s 进行移除 监控上报
            TimeMetric timeMetric = ResponseTimeMetricHolder.getInstance().getTimeMetricMap().get(serviceAddressNode);
            if (timeMetric != null) {
                Duration between = Duration.between(timeMetric.getLastRequestTime(), LocalDateTime.now());
                try {
                    // 响应时间超过5s 该服务节点 清零或失效
                    if (between.getSeconds() > 5) {
                        registerCenter.getCuratorFramework().setData().forPath(
                                Constant.ZkConfig.REGISTER_ROOT + Constant.ZkConfig.SPILT + timeMetric.getClassName() + Constant.ZkConfig.SPILT + serviceAddressNode,
                                "".getBytes());
                        ResponseTimeMetricHolder.getInstance().getTimeMetricMap().clear();

                    } else {
                        // 更新对应service 的 服务 节点时间
                        registerCenter.getCuratorFramework().setData().forPath(
                                Constant.ZkConfig.REGISTER_ROOT + Constant.ZkConfig.SPILT + timeMetric.getClassName() + Constant.ZkConfig.SPILT + serviceAddressNode,
                                timeMetric.getLastTimeConsuming().toString().getBytes());
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        },0,5, TimeUnit.SECONDS);
    }

}
